package br.ufmg.dcc.vod.remoteworkers.async;

import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public abstract class AbstractQueueBasedAsyncEntity<T> extends UnicastRemoteObject {

	private static final long serialVersionUID = 1L;
	
	private final int UNSTARTED = 0;
	private final int RUNNING = 1;
	private final int STOP_NO_FLUSH = 2;
	private final int STOP_AND_FLUSH = 3;
	
	private final CommandQueue<T> queue;
	private final ExecutorService executor;
	private final Runner runner;

	public AbstractQueueBasedAsyncEntity(int port) throws RemoteException {
		super(port);
		this.queue = new CommandQueue<T>();
		this.executor = Executors.newSingleThreadExecutor();
		this.runner = new Runner();
	}
	
	protected synchronized final void queue(T t) {
		if (runner.control.get() == STOP_AND_FLUSH || runner.control.get() == STOP_NO_FLUSH) {
			throw new RuntimeException("This entity has ended serving requests");
		}
		
		if (t == null) { //null is not allowed and used for control purposes
			throw new NullPointerException();
		}
		
		this.queue.put(t);
	}

	public final synchronized void start() {
		if (runner.control.get() != UNSTARTED) {
			throw new RuntimeException("Already running of terminated");
		}

		runner.control.set(RUNNING);
		executor.execute(runner);
	}
	
	public final synchronized void stop(boolean flushQueue, long timeOut, TimeUnit unit) throws InterruptedException {
		if (runner.control.get() != RUNNING) {
			throw new RuntimeException("Not running or already stopped");
		}

		if (flushQueue) {
			runner.control.set(STOP_AND_FLUSH);
		} else {
			runner.control.set(STOP_NO_FLUSH);
		}
		queue.put(null); //null is allowed for control purposes. this call must be after the runner control state is changed
		
		executor.shutdown();
		executor.awaitTermination(timeOut, unit);
	}

	protected abstract void receive(T poll);
	
	private class Runner implements Runnable {

		private AtomicInteger control;

		public Runner() {
			control = new AtomicInteger(UNSTARTED);
		}
		
		@Override
		public void run() {
			while(control.get() == RUNNING) {
				T poll = queue.blockingRemove();
				if (poll != null) {
					receive(poll);
				}
			}
			
			if (control.get() == STOP_AND_FLUSH) {
				while (queue.size() != 0) {
					T poll = queue.blockingRemove();
					if (poll != null) {
						receive(poll);	
					}
				}
			}
		}
	}
}